package chen.bupt.mapreduce.board.feature;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.text.StopWordUtil;
import chen.bupt.text.TextClear;
import chen.bupt.text.TextUtils;
import chen.bupt.util.HDFSFileUtils;

public class BoardFeatureJob {
	private static String inputBaseDir = Constants.BOARD_DATA;
	private static String tmpPath = "/user/chenlingpeng/smth/tmp";
	private static String outputPath = "/user/chenlingpeng/smth/board_feature/";

	public static class Map1 extends
			Mapper<LongWritable, Text, Text, IntWritable> {
		private static final IntWritable one = new IntWritable(1);

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context, "-r-"))
				return;
			String str = value.toString();
			str = TextUtils.splitTextToCont(str, ";", 6)[1];
			if (null == str || "".equals(str))
				return;
			String[] values = str.split(" ");
			for (String feature : values) {
				if ("".equals(feature)) {
					continue;
				}

				if (StopWordUtil.isSpecialTag(feature) || feature == null)
					continue;
				context.write(new Text(feature), one);
			}
		}
	}

	public static class Reduce1 extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {

			int sum = 0;
			for (IntWritable value : values) {
				sum += value.get();
			}
			String keyS = key.toString();

			if (StopWordUtil.isSpecialTag(keyS)) {
				return;
			}
			context.write(new Text(keyS), new IntWritable(sum));
		}
	}

	public static class Map2 extends Mapper<LongWritable, Text, Text, Text> {
		private static final Text one = new Text("key");

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			context.write(one, value);
		}
	}

	public static class Reduce2 extends Reducer<Text, Text, Text, NullWritable> {
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			StringBuffer sb = new StringBuffer();
			int count = 0;
			for (Text val : values) {
				String[] kv = val.toString().split("\t");
				if (kv[0].contains("/nr")) {
					if (Integer.parseInt(kv[1]) < 10) {
						continue;
					} else {
						kv[0] = TextClear.substring(kv[0], "/nr");
					}
				}
				count++;
				sb.append(kv[0] + ":" + kv[1] + ";");
			}
			context.write(new Text(sb.toString()), NullWritable.get());
			context.write(new Text("total feature: " + count), NullWritable
					.get());
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration confg = new Configuration();
		confg.set("fs.default.name", "hdfs://10.1.1.26:31054");
		FileSystem fs = FileSystem.get(confg);
		FileStatus[] fsDirs = fs.listStatus(new Path(inputBaseDir));
		for (FileStatus fss : fsDirs) {
			if (!fss.isDir())
				continue;
			Path dirName = fss.getPath();
			String boardId = dirName.getName();
			if (boardId.startsWith("_"))
				continue;
			if (!boardId.startsWith(args[0]))
				continue;
			Configuration conf = new Configuration();
			Job job = new Job(conf, "BoardFeature");
			job.setJarByClass(BoardFeatureJob.class);
			job.setMapperClass(Map1.class);
			job.setCombinerClass(Reduce1.class);
			job.setReducerClass(Reduce1.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);
			Path input = new Path(inputBaseDir + "/" + boardId);
			Path tmp = new Path(tmpPath+boardId);
			Path output = new Path(outputPath + boardId);
			HDFSFileUtils.deleteFile(output, conf);
			HDFSFileUtils.deleteFile(tmp, conf);
			FileInputFormat.addInputPath(job, input);
			FileOutputFormat.setOutputPath(job, tmp);
			job.waitForCompletion(true);

			Configuration conf2 = new Configuration();
			Job job2 = new Job(conf2, "BoardFeature");
			job2.setJarByClass(BoardFeatureJob.class);
			job2.setMapperClass(Map2.class);
			job2.setReducerClass(Reduce2.class);
			job2.setNumReduceTasks(1);
			job2.setOutputKeyClass(Text.class);
			job2.setOutputValueClass(Text.class);
			FileInputFormat.addInputPath(job2, tmp);
			FileOutputFormat.setOutputPath(job2, output);
			job2.waitForCompletion(true);
		}
	}
}
